
#include <zmq.hpp>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <stdint.h>
#include <iostream>
#include <thread>

#define within(num) (rand() % (num - num / 10 + 1)) + num / 10

/*
    id length must be 5
*/

const char *id_A = "1111A";
const char *id_B = "1111B";
const char *id_C = "1111C";

static uint64_t s_index = 0;

void pubMsg(const char *id, const std::string &msg, zmq::socket_t &publisher)
{
    std::string index = ", index: " + std::to_string(s_index++);

    zmq::message_t zmq_msg(msg.size() + index.size() + 5);
    memcpy((char *)zmq_msg.data(), id, 5);
    memcpy((char *)zmq_msg.data() + 5, msg.c_str(), msg.size());
    memcpy((char *)zmq_msg.data() + 5 + msg.size(), index.c_str(), index.size());
    publisher.send(zmq_msg);
}

void publish()
{
    // Prepare context and publisher
    zmq::context_t context(1);
    zmq::socket_t publisher(context, ZMQ_PUB);
    std::string url = "ipc://server.ipc";
    publisher.bind(url); // Not usable on Windows.
    std::cout << "publish url: " << url << std::endl;

    srandom((unsigned)time(NULL));
    while (1)
    {
        int code = within(100);
        if (code == 10)
            pubMsg(id_A, "hello A", publisher);
        else if (code == 11)
            pubMsg(id_B, "hello B", publisher);
        else
            pubMsg(id_C, "hello C", publisher);
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
}

void doRecv(const char *id, const zmq::message_t &msg)
{
    std::string data = std::string((char *)msg.data(), msg.size());
    std::cout << "id: " << id << ", data: " << data << std::endl;
}

void subcribe(const char *id)
{
    std::string url = "ipc://" + std::string(id) + ".ipc";
    zmq::context_t context(1);
    zmq::socket_t subscriber(context, ZMQ_SUB);
    subscriber.connect(url);
    subscriber.setsockopt(ZMQ_SUBSCRIBE, "", 0);
    std::cout << "subcribe url: " << url << std::endl;

    while (1)
    {
        zmq::message_t msg;
        subscriber.recv(&msg);
        doRecv(id, msg);
    }
}

int main()
{
    std::thread sendThread(publish);
    sendThread.detach();

    std::thread recvThreadA(subcribe, id_A);
    recvThreadA.detach();

    std::thread recvThreadB(subcribe, id_B);
    recvThreadB.detach();

    while (true)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    return 0;
}
